package com.hexiang.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * 基于sql过滤的消息消费监听器
 */
@Slf4j
@Component
public class SqlFilterConsumer {
    private String consumerGroup = "sql-filter-topic-consumer-group";
    private String namesrvAddr = "127.0.0.1:9876";
    private String topicName = "test_sql_filter_topic";

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = consumerInit();
        consumer.start();
    }

    @PreDestroy
    public void shutdown() {
        consumer.shutdown();
    }

    private DefaultMQPushConsumer consumerInit() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        // 订阅主题和 标签
        consumer.subscribe(topicName,
                MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
                        "and (a is not null and a between 0 and 3)"));

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            try {
                for (Message msg : msgs) {
                    // 消费者获取消息 这里只输出 不做后面逻辑处理
                    String body = new String(msg.getBody(), "UTF-8");
                    log.info("  >>> [消费端] 获取消息主题 topic :: {}, tags :: {}, 消费消息 body :: {} ", msg.getTopic(), msg.getTags(), body);
                }
            } catch (UnsupportedEncodingException e) {
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        return consumer;
    }

}